package main

import (
	"flag"
	"fmt"
	"github.com/Shopify/sarama"
	"time"
)

var tNow1 = time.Now()

var count1 = 0
var c1 = XLConsumer{}

var last = time.Now()
var Cnt = 0

func DataCB(msg *sarama.ConsumerMessage) {
	count1++
	Cnt++
	c1.MarkOffset(msg)
	fmt.Printf("Topic:%v, offest:%v\n", msg.Topic, msg.Offset)
	if time.Since(tNow1) > time.Second*3 {
		fmt.Println("总消费条数:", count1*1000/(int(time.Since(tNow1).Nanoseconds())/1000000))
		count1 = 0
		tNow1 = time.Now()
	}
	if Cnt%100 == 0 {
		fmt.Printf("消费%v条数据，总共耗时:%v\n", Cnt, time.Now().Sub(last))
	}
	if Cnt == 20000 {
		fmt.Printf("消费%v条数据，总共耗时:%v\n", Cnt, time.Now().Sub(last))
		Cnt = 0
		last = time.Now()
	}
}

func main() {
	brokers := flag.String("broker", "172.20.32.239:9092", "help message for brokers")
	topic := flag.String("topic", "snappy", "help message for topic")
	Offset := flag.String("offset", "Oldest", "help message for Oldest/Newest")
	GroupID := flag.String("groupID", "consumer_group_1", "help message for GroupID")
	flag.Parse()
	fmt.Printf("%v\n", *brokers)
	fmt.Printf("%v\n", *topic)
	fmt.Printf("%v\n", *Offset)

	conf := KafkaCfg{}
	conf.Consumer.Brokers = *brokers
	conf.Consumer.Topics = *topic
	conf.Consumer.GroupID = *GroupID
	conf.Consumer.InitOffset = *Offset

	fmt.Printf("%v\n", conf)
	err := c1.Init(&conf, DataCB)
	if err != nil {
		fmt.Println(err)
		return
	}

	for {
		time.Sleep(time.Second)
	}

	err = c1.Fini()
	if err != nil {
		fmt.Println(err)
		return
	}
}
